package cn.edu.ahstu.flink.java;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class WindowWordCount {

    /**
     * 主函数，用于执行基于窗口的词频统计
     *
     * @param args 命令行参数，本程序未使用
     * @throws Exception 如果执行流处理环境中的作业时发生错误，则会抛出异常
     */
    public static void main(String[] args) throws Exception {

        // 获取流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 从指定的socket地址和端口读取数据流，并进行分割处理，然后按第一个字段分组，以5秒为窗口进行词频统计
        DataStream<Tuple2<String, Integer>> dataStream = env
                .socketTextStream("hadoop1", 9999)
                .flatMap(new Splitter())
                .keyBy(0)
                .timeWindow(Time.seconds(5))
                .sum(1);

        // 将统计结果打印到控制台
        dataStream.print();

        // 执行流处理环境中的作业
        env.execute("Window WordCount");
    }


    public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        /**
         * 将输入的句子拆分成单词，并对每个单词生成一个二元组，其中包含单词本身和数字1
         * 这个方法用于实现扁平化映射操作，通常是用于MapReduce编程模型中的映射步骤
         * 它将句子中的每个单词映射为一个键值对，键是单词本身，值是数字1，用于后续的归约操作
         *
         * @param sentence 输入的句子，将被拆分成单词
         * @param out      收集器对象，用于收集生成的二元组
         * @throws Exception 如果处理过程中遇到异常，将向上抛出
         */
        @Override
        public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {
            // 使用正则表达式分割句子中的单词，以空格为分隔符
            for (String word : sentence.split(" ")) {
                // 对于每个单词，创建一个二元组，并将单词本身作为键，数字1作为值
                // 然后通过收集器对象将这个二元组输出
                out.collect(new Tuple2<String, Integer>(word, 1));
            }
        }
    }
}
